package com.sh.data.engine.domain.shims.hbase.util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.sh.data.engine.domain.base.model.entity.PreviewDataInfoEntity;
import com.sh.data.engine.domain.shims.kafka.util.SocketUtil;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.*;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/11 16:38
 */
public class HBaseUtil {

    private static final String ARG_ERROR_MSG = "hbase.zookeeper.quorum参数未填写或填写错误，请检查";

    private static final int DEFAULT_SOCKET_TIMEOUT = 1;

    public static PreviewDataInfoEntity previewData(
        Connection conn, String tableName, boolean nullAsEmpty)
        throws IOException, IllegalArgumentException {
        if (StringUtils.isBlank(tableName)) {
            return null;
        }
        int limit = 10;
        List<Map<String, String>> mapList = Lists.newArrayList();
        Set<String> columns = new TreeSet<>();
        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Scan scan = new Scan();
            scan.setCaching(limit);
            ResultScanner scanner = table.getScanner(scan);
            Iterator<Result> res = scanner.iterator();
            int count = 0;
            while (res.hasNext() && count < limit) {
                Map<String, String> rowData = new HashMap<String, String>();
                Result result = res.next();
                for (Cell cell : result.rawCells()) {
                    String rowKey = getRowKey(cell);
                    String family = getFamily(cell);
                    String column = getColumn(cell);
                    String value = getValue(cell);
                    String key = StringUtils.isBlank(family) ? column : (family + ":" + column);
                    columns.add(key);
                    rowData.put(key, value);
                    rowData.put("row", rowKey);
                }
                mapList.add(rowData);
                count++;
            }
        }

        if (CollectionUtils.isEmpty(mapList)) {
            return PreviewDataInfoEntity.builder()
                .fieldNameList(Collections.emptyList())
                .dataList(Collections.emptyList())
                .build();
        }

        List<String> fieldNameList = new ArrayList<>();
        fieldNameList.add("row");
        fieldNameList.addAll(columns);

        List<List<Object>> valuesList = new LinkedList<>();
        for (Map<String, String> map : mapList) {
            List<Object> value = new LinkedList<>();
            for (String column : fieldNameList) {
                if (nullAsEmpty) {
                    value.add(map.getOrDefault(column, ""));
                } else {
                    value.add(map.get(column));
                }
            }
            valuesList.add(value);
        }

        return PreviewDataInfoEntity.builder()
            .fieldNameList(fieldNameList)
            .dataList(valuesList)
            .build();
    }

    public static List<String> listCollectionNames(Connection conn)
        throws IOException, IllegalArgumentException {
        List<String> list = new LinkedList<>();
        try (Admin admin = conn.getAdmin()) {
            TableName[] tableNames = admin.listTableNames();
            for (TableName tableName : tableNames) {
                list.add(tableName.getNameAsString());
            }
        }
        return list;
    }

    public static List<String> getColumnFamilyNames(Connection conn, String tableName)
        throws Exception {
        List<String> result = new ArrayList<>();

        try (Table table = conn.getTable(TableName.valueOf(tableName))) {
            Set<byte[]> nameByteArrList = table.getDescriptor().getColumnFamilyNames();

            for (byte[] nameByteArr : nameByteArrList) {
                result.add(new String(nameByteArr, StandardCharsets.UTF_8));
            }
        }

        return result;
    }

    public static void testConnection(String configJson)
        throws IOException, IllegalArgumentException {
        try (Connection conn = getConnection(configJson)) {
        }
    }

    public static Connection getConnection(String configJson)
        throws IOException, IllegalArgumentException {
        if (StringUtils.isBlank(configJson)) {
            throw new IllegalArgumentException(ARG_ERROR_MSG);
        }

        JSONObject jsonObject = null;
        try {
            jsonObject = JSON.parseObject(configJson);
        } catch (JSONException e) {
            throw new IllegalArgumentException(ARG_ERROR_MSG);
        }

        boolean foundZookeeper = false;
        Configuration configuration = HBaseConfiguration.create();
        for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
            String key = StringUtils.trim(entry.getKey());
            String value = StringUtils.trim(Objects.toString(entry.getValue().toString()));
            if (StringUtils.equalsIgnoreCase(key, "hbase.zookeeper.quorum")) {
                foundZookeeper = true;
                testSocketConnection(value, DEFAULT_SOCKET_TIMEOUT);
            }
            configuration.set(key, value);
        }

        if (!foundZookeeper) {
            throw new IllegalArgumentException(ARG_ERROR_MSG);
        }

        configuration.set("hbase.client.operation.timeout", "1000");
        configuration.set("hbase.rpc.timeout", "1000");
        configuration.set("hbase.client.retries.number", "0");
        configuration.set("zookeeper.session.timeout", "1000");
        configuration.set("zookeeper.recovery.retry", "0");
        configuration.set("zookeeper.recovery.retry.intervalmill", "0");
        return ConnectionFactory.createConnection(configuration);
    }

    private static void testSocketConnection(String address, Integer timeout)
        throws IOException, IllegalArgumentException, SecurityException {
        try {
            List<Pair<String, Integer>> list = SocketUtil.parseHostAndPort(address, 2181);
            SocketUtil.testSocketConnection(list, timeout);
        } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException(ARG_ERROR_MSG, e);
        }
    }

    private static String getRowKey(Cell cell) {
        return Bytes.toStringBinary(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
    }

    private static String getFamily(Cell cell) {
        return Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
    }

    private static String getColumn(Cell cell) {
        return Bytes.toString(
            cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
    }

    private static String getValue(Cell cell) {
        return Bytes.toStringBinary(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
    }

  /*public static void main(String... args) throws Exception {
    String configJson = "{\"hbase.zookeeper.quorum\": \"10.88.36.79:2181\"}";
    testConnection(configJson);
    Connection conn = getConnection(configJson);
    listCollectionNames(conn).forEach(System.out::println);
    System.out.println(previewData(conn, "t4", true));
  }*/
}
